package com.walker.springboot.rabbitmq.config;

import com.sun.org.apache.xpath.internal.operations.Or;
import com.walker.springboot.rabbitmq.listener.MyListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description:
 * @ProjectName: walker-step
 * @Package: com.walker.springboot.rabbitmq.config
 * @ClassName: RabbitmqConfig
 * @Author: yangyl
 * @CreateDate: 2020/6/21 20:23
 */
@Configuration
public class RabbitmqConfig {

    private Logger logger = LoggerFactory.getLogger(RabbitmqConfig.class);

    @Autowired
    protected Environment environment;
    @Autowired
    protected CachingConnectionFactory connectionFactory;
    @Autowired
    protected SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    /**
     * 单一消费者
     *
     * @return
     */
    @Bean("singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        return factory;
    }

    /**
     * 多个消费者
     *
     * @return
     */
    @Bean("multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setConcurrentConsumers(environment.getProperty("spring.rabbitmq.listener.simple.concurrency", int.class));
        factory.setMaxConcurrentConsumers(environment.getProperty("spring.rabbitmq.listener.simple.max-concurrency", int.class));
        factory.setPrefetchCount(environment.getProperty("spring.rabbitmq.listener.simple.prefetch", int.class));
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                logger.info("消息发送成功:{}--{},{}", correlationData, ack, cause);
            }
        });
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                logger.info("消息丢失:exchange:{},routingKey{},replyCode:{},replyText:{},message:{}", exchange, routingKey, replyCode, replyText, message);
            }
        });
        return rabbitTemplate;
    }

//
//    @Bean
//    public DirectExchange basicExchange() {
//        return new DirectExchange(environment.getProperty("rabbitmq.config.basic.exchange-key"), true, false);
//    }
//
//    @Bean
//    public Queue basicQueue() {
//        return new Queue(environment.getProperty("rabbitmq.config.basic.queue-key"), true);
//    }
//
//    @Bean
//    public Binding basicBinding() {
//        return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(environment.getProperty("rabbitmq.config.basic.routing-key"));
//    }



    @Bean(name = "simpleMessageListenerContainer")
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer simpleContainer = new SimpleMessageListenerContainer();
        simpleContainer.setConnectionFactory(connectionFactory);
        simpleContainer.setMaxConcurrentConsumers(10);
        simpleContainer.setConcurrentConsumers(5);
        simpleContainer.setPrefetchCount(5);
        simpleContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return simpleContainer;
    }

    @Bean
    public Queue ttlQueue(){
        Map<String, Object> args=new HashMap<String, Object>(2);
        args.put("x-message-ttl",3000);  //过期时间
        args.put("x-dead-letter-exchange","waring.exchange.03"); //死信交换机
        args.put("x-dead-letter-exchange",""); //死信交换机
        Queue queue=new Queue("ttl.queue",true,false,false,args);
        return queue;
    }
    @Bean
    public DirectExchange ttlExchange(){
        return new DirectExchange("ttl.exchange", true, false);
    }

    @Bean
    public Binding ttlQueueBinding() {
        return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("xx");
    }


}
